package com.atguigu.app.dwd.db;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTestTestPaper {
    public static void main(String[] args) {
        //1、获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //设置并行度为1
        env.setParallelism(1);

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        // TODO 3. 从 Kafka 读取业务数据，封装为 Flink SQL 表
        tableEnv.executeSql("create table topic_db(" +
                "`database` String,\n" +
                "`table` String,\n" +
                "`type` String,\n" +
                "`data` map<String, String>,\n" +
                "`old` map<String, String>,\n" +
                "`pt` as PROCTIME(),\n" +
                "`ts` string\n" +
                ")" + KafkaUtil.getKafkaDDL("topic_db", "dwd_test_test_paper"));

        // TODO 4 过滤出订单详情表
        Table testExam = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['paper_id'] paper_id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['score'] score,\n" +
                "  `data`['duration_sec'] duration_sec,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `data`['submit_time'] submit_time,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  `data`['deleted'] deleted,\n" +
                "  `pt`,\n" +
                "  `ts`\n" +
                "from topic_db\n" +
                "where `table`='test_exam'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("test_exam",testExam);



        Table testExamQuestion = tableEnv.sqlQuery("select\n" +
                "  `data`['id'] id,\n" +
                "  `data`['exam_id'] exam_id,\n" +
                "  `data`['paper_id'] paper_id,\n" +
                "  `data`['question_id'] question_id,\n" +
                "  `data`['user_id'] user_id,\n" +
                "  `data`['answer'] answer,\n" +
                "  `data`['is_correct'] is_correct,\n" +
                "  `data`['score'] score,\n" +
                "  `data`['create_time'] create_time,\n" +
                "  `data`['update_time'] update_time,\n" +
                "  `data`['deleted'] deleted,\n" +
                "  `pt`,\n" +
                "  `ts`\n" +
                "from topic_db\n" +
                "where `table`='test_exam_question'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("test_exam_question",testExamQuestion);

        Table resultTable = tableEnv.sqlQuery("select\n" +
                "  te.id,\n" +
                "  te.paper_id,\n" +
                "  te.user_id,\n" +
                "  teq.question_id," +
                "  te.score scoreSum,\n" +
                "  teq.score score,\n" +
                "  te.duration_sec,\n" +
                "  te.create_time,\n" +
                "  te.update_time,\n" +
                "  teq.exam_id,\n" +
                "  teq.answer,\n" +
                "  teq.is_correct,\n" +
                "  te.ts\n" +
                "  from\n" +
                "  test_exam te\n" +
                "  join\n" +
                "  test_exam_question teq\n" +
                "  on te.paper_id=teq.paper_id and te.user_id=teq.user_id");

        tableEnv.createTemporaryView("result_table",resultTable);


        tableEnv.executeSql("create table dwd_test_test_exam(\n" +
                "   id string,\n" +
                "   paper_id string,\n" +
                "   user_id string,\n" +
                "   question_id string," +
                "   scoreSum String,\n" +
                "   score string,\n" +
                "   duration_sec string,\n" +
                "   create_time string,\n" +
                "   update_time string,\n" +
                "   exam_id string,\n" +
                "   answer string,\n" +
                "   is_correct string,\n" +
                "   ts string \n" +
                ")" + KafkaUtil.getKafkaSinkDDL("dwd_test_test_exam"));



        tableEnv.executeSql("insert into dwd_test_test_exam select * from result_table");




    }
}
